#pragma once 

/* 实现消息转发子服务部分的封装 */

#include <brpc/server.h>
#include <butil/logging.h>
#include <odb/database.hxx>
#include <odb/mysql/database.hxx>
#include <brpc/channel.h>

#include "brpc.hpp"     //Rpc调用模块的封装
#include "etcd.hpp"     //服务注册模块封装
#include "logger.hpp"   //日志模块封装
#include "mysql_chat_session_member.hpp"    //会话表的ODB映射操作
#include "mysql_user.hpp"   //用户表的ODB映射操作
#include "utils.hpp"    //工具类接口
#include "rabbitmq.hpp" //RabbitMQ的二次封装模块

#include "user.pb.h"    //protobuf框架代码
#include "base.pb.h"    //protobuf框架代码
#include "transmite.pb.h"    //protobuf框架代码

#include "user.hxx"     //ODB框架代码
#include "user-odb.hxx" //ODB框架代码
#include "chat_session_member.hxx"     //ODB框架代码
#include "chat_session_member.hxx" //ODB框架代码


namespace ChatSystem
{

    // 1. 创建rpc服务子类继承pb(protobuf)中的MsgTransmitService服务类，并实现内部的业务接口逻辑；
    class MsgTransmiteServiceImpl : public ChatSystem::MsgTransmiteService
    {
    public:
        MsgTransmiteServiceImpl(const std::shared_ptr<odb::core::database>& mysql_db,
            const ServiceManager::ptr& channel_manager,
            const MQClient::ptr& mq_client,
            const std::string& user_service_name, 
            const std::string& exchange_name, 
            const std::string& routing_key)
            :_mysql_session_member_table(std::make_shared<ChatSessionMemberTable>(mysql_db))
            ,_channel_manager(channel_manager)
            ,_mq_client(mq_client)
            ,_user_service_name(user_service_name)
            ,_exchange_name(exchange_name)
            ,_routing_key(routing_key)
        {}

        ~MsgTransmiteServiceImpl(){}

        //获取消息转发列表接口
        void GetTransmiteTarget(google::protobuf::RpcController* controller,
            const ChatSystem::NewMessageReq* request,
            ChatSystem::GetTransmiteTargetRsp* response,
            google::protobuf::Closure* done) override
        {
            LOG_DEBUG("收到获取消息转发列表的请求！");
            //使用ClosureGurd类管理done对象，防止出现未调用Run方法；
            brpc::ClosureGuard rpc_gurd(done);
            //定义一个错误处理函数，当出错的时候被调用
            auto err_response = [this, response](const std::string& errmsg)
            {
                response->set_success(false);
                response->set_errmsg(errmsg);
                return;
            };

            response->set_request_id(request->request_id());
            
            //1. 从请求中取出消息内容，会话ID，用户ID
            std::string uid = request->user_id();
            std::string chat_session_id = request->chat_session_id();
            const MessageContent& msg_content = request->message();

            //2. 根据用户 ID 从用户子服务获取当前发送者用户信息
            auto channel = _channel_manager->choose(_user_service_name);
            if(!channel)
            {
                LOG_ERROR("{} - 未找到用户管理子服务节点 - {}!", request->request_id(), _user_service_name);
                return err_response("未找到用户管理子服务节点!");
            }
            UserService_Stub stub(channel.get());
            GetUserInfoReq req;
            GetUserInfoRsp rsp;
            req.set_request_id(request->request_id());
            req.set_user_id(uid);
            req.set_login_session_id(request->login_session_id());
            brpc::Controller cntl;
            stub.GetUserInfo(&cntl, &req, &rsp, nullptr);
            if(cntl.Failed() || rsp.success() == false)
            {
                LOG_ERROR("{} - 用户管理子服务 {} Rpc调用失败, {}!", request->request_id(), _user_service_name, cntl.ErrorText());
                return err_response("用户管理子服务Rpc调用失败!");
            }

            //3. 根据消息内容构造完成的消息结构（分配消息ID，填充发送者信息，填充消息产生时间）
            MessageInfo message;
            message.set_message_id(Util::uuid());
            message.set_chat_session_id(chat_session_id);
            message.set_timestamp(time(nullptr));
            message.mutable_sender()->CopyFrom(rsp.user_info());
            message.mutable_message()->CopyFrom(msg_content);
            //4. 将消息序列化后发布到 MQ 消息队列中，让消息存储子服务对消息进行持久化存储
            bool ret = _mq_client->publish(_exchange_name, message.SerializeAsString(), _routing_key);
            if(ret == false)
            {
                LOG_ERROR("{} - 消息队列发布消息失败!", request->request_id());
                return err_response("消息队列发布消息失败!");
            }
            //5. 从数据库获取目标会话所有成员 ID
            auto target_session_members = _mysql_session_member_table->members(chat_session_id);
            //6. 组织响应（完整消息+目标用户 ID），发送给网关，告知网关该将消息发送给谁
            response->set_success(true);
            response->mutable_message()->CopyFrom(message);
            for(const auto& id : target_session_members)
            {
                response->add_target_id_list(id);
            }
        }

    private:
        /*1. 进行消息的组织：需要发送者(通过用户子服务获取信息)、所属会话(请求中的内容)、消息内容(请求中的内容)、
        产生时间(这个需要根据服务器的当前时间来生成)、消息ID(这个需要服务器自己生成);
        2. 获取消息转发客户端用户列表；
        3. 将封装完毕的消息，发布到消息队列中，待消息存储子服务进行消息持久化 */
        ChatSystem::UserTable::ptr _mysql_user;     //MySQL的user表， 一定是先实例化odb操作句柄，在实例化这个，注意这个顺序；
        ServiceManager::ptr _channel_manager;    //Rpc调用中信道管理对象
        // ChatSystem::Discovery::ptr _service_discoverer; //服务发现对象
        //聊天会话成员表操作句柄
        ChatSessionMemberTable::ptr _mysql_session_member_table;
        //消息队列操作句柄
        ChatSystem::MQClient::ptr _mq_client;
        std::string _user_service_name; //用户子服务调用相关信息
        std::string _exchange_name; //RabbitMQ的交换机名称
        std::string _routing_key;   //RabbitMQ中路由消息的key
    };



    //用户管理子服务器搭建的流程类
    class MsgTransmiteServer
    {
    public:
        using ptr = std::shared_ptr<MsgTransmiteServer>;

        MsgTransmiteServer(const Discovery::ptr& discovery_client, 
            const Registry::ptr& registry_client, 
            const std::shared_ptr<odb::core::database>& mysql_client,
            const std::shared_ptr<brpc::Server>& rpc_server)
        :_discovery_client(discovery_client)
        , _registry_client(registry_client)
        ,_mysql_client(mysql_client)
        , _rpc_server(rpc_server)
        {}
        ~MsgTransmiteServer(){}

        //搭建RPC服务器，并启动服务器
        /* make相关接口的调用必须在start接口调用之前，
        因为start涉及到了启动服务器，而启动服务器的前提是客户端对象和服务注册客户端对象已经实例化好了; 
        为了防止使用者在start接口调用之前未调用 make相关接口而导致
        服务器无法启动的问题，我们可以用建造者模式，将构造过程进行控制，在代码中规定调用顺序，去倒逼使用者;
        使用建造者模式的原因还是由于构造userServer对象时参数太多，为了简化userServer对象的构造 */
        void start()
        {
            // 阻塞等待服务端运行结束
            _rpc_server->RunUntilAskedToQuit();
        }
    private:
        Registry::ptr _registry_client;  //注册中心客户端
        Discovery::ptr _discovery_client;    //服务发现客户端
        std::shared_ptr<odb::core::database> _mysql_client;    //MySQL的ODB操作句柄
        std::shared_ptr<brpc::Server> _rpc_server;   //brpc服务器
    };

    //建造者模式---用来构造UserServer对象，将UserServer对象的真正构造过程给封装起来，通过封装的类去构造 UserServer对象
    class MsgTransmiteServerBuilder
    {
    public:
        //构造mysql客户端对象
        void make_mysql_object(const std::string& user,
            const std::string& password,
            const std::string& host,
            const std::string& db,
            const std::string& cset,
            int port,
            int connection_pool_count)
        {
           _mysql_client = ODBFactory::create(user, password, host, db, cset, port, connection_pool_count);
        }

        //构造RabbitMQ客户端对象
        void make_mq_object(const std::string& user, 
        const std::string& password, 
        const std::string& host, 
        const std::string& exchange_name,   //交换机名称
        const std::string& queue_name,      //消息队列名称
        const std::string& routing_key)      
        {
            _exchange_name = exchange_name;
            _routing_key = routing_key;
           _mq_client = std::make_shared<ChatSystem::MQClient>(user, password, host);
           _mq_client->declareComponents(exchange_name, queue_name, routing_key);
        }

        //构造服务发现客户端对象和信道管理对象
        void make_discovery_object(const std::string& reg_host,
            const std::string& base_service_name,
            const std::string& user_service_name)
        {
            _user_service_name = user_service_name;
            _channel_manager = std::make_shared<ServiceManager>();
            _channel_manager->declared(_user_service_name);
            LOG_DEBUG("设置文件子服务为需要添加管理的子服务：{}", _user_service_name);
            auto put_cb = std::bind(&ChatSystem::ServiceManager::onServiceOnline, 
                _channel_manager.get(), 
                std::placeholders::_1, 
                std::placeholders::_2);
            auto del_cb = std::bind(&ChatSystem::ServiceManager::onServiceOffline, 
                _channel_manager.get(), 
                std::placeholders::_1,
                std::placeholders::_2);
            //2. 构造服务发现对象 
            _discovery_client = std::make_shared<ChatSystem::Discovery>(reg_host, base_service_name, put_cb, del_cb);
        }

        //用于构建RPC服务器
        void make_rpc_server(uint16_t port, 
            uint32_t timeout,      //连接空闲超时时间---超时后连接关闭
            uint8_t num_threads)    //rpc服务器的线程数量---IO线程数量；
        {
            _rpc_server = std::make_shared<brpc::Server>();

            if(!_mysql_client)
            {
                LOG_ERROR("还未初始化MySQL数据库模块！");
                abort();
            }
            if(!_mq_client)
            {
                LOG_ERROR("还未初始化消息队列客户端模块！");
                abort();
            }
            if(!_channel_manager)
            {
                LOG_ERROR("还未初始化信道管理模块！");
                abort();
            }
            if(!_discovery_client)
            {
                LOG_ERROR("还未初始化服务发现模块！");
                abort();
            }

            // 向rpc服务器类中添加rpc子服务对象---告诉服务器收到什么请求用哪个接口处理；
            /*注意：如果在msg_transmite_service对象是栈上创建的，AddService的第一个参数就是msg_transmite_service对象的地址，当
            make_rpc_server函数调用结束后msg_transmite_service这个对象是被销毁了的，而在 _rpc_server中会使用到 msg_transmite_service对象
            此时就会出现段错误，所以我们这里需要在堆上创建对象，这样make_rpc_server函数调用结束后再使用user_service对象也不会错。*/
            MsgTransmiteServiceImpl* msg_transmite_service = new MsgTransmiteServiceImpl(_mysql_client, 
                _channel_manager,
                _mq_client, 
                _user_service_name, 
                _exchange_name, 
                _routing_key);

            /* SERVER_DOESNT_OWN_SERVICE 表示添加服务失败时，服务器也不会删除服务器对象； SERVER_OWNS_SERVICE 表示添加服务失败时，
            服务器将负责删除服务对象，当我们在堆上创建user_service对象时需要使用SERVER_OWNS_SERVICE，就不会造成内存泄漏。 */
            int ret = _rpc_server->AddService(msg_transmite_service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE); 

            if(ret == -1)
            {
                LOG_ERROR("添加Rpc服务失败！ ");
                abort();
            }
            brpc::ServerOptions options;
            options.idle_timeout_sec - timeout;
            options.num_threads = num_threads;

            ret = _rpc_server->Start(port, &options);
            if(ret == -1)
            {
                LOG_ERROR("服务器启动失败！");
                abort();
            }
        }

        //用于构造服务注册客户端对象
        void make_registry_object(const std::string& reg_host,  //注册中心的地址
            const std::string& service_name,    //服务名称
            const std::string& access_host)     //提供服务器的主机的地址信息
        {
            if(!_rpc_server)
            {
                LOG_ERROR("还未初始化RPC服务器！");
                abort();
            }
            _registry_client = std::make_shared<Registry>(reg_host);
            _registry_client->registry(service_name, access_host);
        }

        //构造一个 UserServer对象
        MsgTransmiteServer::ptr build()
        {
            if(!_mysql_client)
            {
                LOG_ERROR("还未初始化MySQL数据库模块！");
                abort();
            }
            if(!_channel_manager)
            {
                LOG_ERROR("还未初始化信道管理模块！");
                abort();
            }
            if(!_mq_client)
            {
                LOG_ERROR("还未初始化消息队列客户端模块！");
                abort();
            }
            if(!_discovery_client)
            {
                LOG_ERROR("还未初始化服务发现模块！");
                abort();
            }
            if(!_registry_client)
            {
                LOG_ERROR("还未初始化服务注册模块");
                abort();
            }
            if(!_rpc_server)
            {
                LOG_ERROR("还未初始化RPC服务器模块");
                abort();
            }

            MsgTransmiteServer::ptr server = std::make_shared<MsgTransmiteServer>(_discovery_client, 
                _registry_client, 
                _mysql_client,
                _rpc_server);
            return server;
        }
    private:
        std::string _user_service_name;  //用户管理子服务名称
        std::shared_ptr<odb::core::database> _mysql_client; //MySQL的ODB操作句柄
        ServiceManager::ptr _channel_manager;   //信道管理对象
        Registry::ptr _registry_client;  //注册中心客户端
        Discovery::ptr _discovery_client;    //服务发现客户端对象
        MQClient::ptr _mq_client;   //消息队列客户端
        std::shared_ptr<brpc::Server> _rpc_server;   //brpc服务器
        std::string _exchange_name;     //交换机名称
        std::string _routing_key;   //路由消息的key
    };

}